package com.xinxin.socket.sample.context;

import com.google.common.collect.Maps;
import com.xinxin.socket.constant.CommandId;
import com.xinxin.socket.context.ChannelLoginContext;
import com.xinxin.socket.msg.NetMsg;
import com.xinxin.socket.proto.Messagepush;
import io.netty.channel.Channel;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;
import java.util.Deque;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentMap;

@Slf4j
public class TopicContext {

    private static final String[][] CON_DETAILS = new String[][]{
            new String[]{"Mars", "0", "STN Discuss"},
            new String[]{"Mars", "1", "XLog Discuss"},
            new String[]{"Mars", "2", "SDT Discuss"}
    };

    private static final TopicContext INSTANCE = new TopicContext();

    private final ConcurrentMap<String, Deque<Channel>> topicJoiners = Maps.newConcurrentMap();

    private TopicContext() {
        Arrays.stream(CON_DETAILS).forEach(conDetail -> createTopic(conDetail[1]));
    }

    public static TopicContext getInstance() {
        return INSTANCE;
    }

    public Deque<Channel> createTopic(String topicName) {
        Deque<Channel> topicQueue = new ConcurrentLinkedDeque<>();
        topicJoiners.put(topicName, topicQueue);
        return topicQueue;
    }

    public Deque<Channel> removeTopic(String topicName) {
        return topicJoiners.remove(topicName);
    }

    public Set<String> getTopicSet() {
        return topicJoiners.keySet();
    }

    public Deque<Channel> getTopicQueue(String topicName) {
        Deque<Channel> topicQueue = topicJoiners.get(topicName);
        if (topicQueue == null) {
            log.warn("No topic: {}", topicName);
            return null;
        }

        return topicQueue;
    }

    public void joinTopic(Channel ch) {
        topicJoiners.keySet().forEach(topicName -> joinTopic(topicName, ch));
    }

    public void joinTopic(String topicName, Channel ch) {
        Optional.ofNullable(getTopicQueue(topicName)).ifPresent(topicQueue -> {
            if (topicQueue.contains(ch)) {
                log.info("Channel {} already in topic: {}", ch.id(), topicName);
                return;
            }

            topicQueue.offer(ch);
        });
    }

    public void leftTopic(Channel ch) {
        topicJoiners.keySet().forEach(topicName -> leftTopic(topicName, ch));
    }

    public void leftTopic(String topicName, Channel ch) {
        Optional.ofNullable(getTopicQueue(topicName)).ifPresent(topicQueue -> topicQueue.remove(ch));
    }

    public void pushMessage(String topicName, String content, Channel ch) {
        new Pusher(topicName, content, ch).start();
    }

    @AllArgsConstructor
    private class Pusher extends Thread {

        private String topicName;
        private String content;
        private Channel ch;

        @Override
        public void run() {
            Optional.ofNullable(getTopicQueue(topicName)).ifPresent(topicQueue -> {
                Messagepush.MessagePush messagePush = Messagepush.MessagePush.newBuilder()
                        .setTopic(topicName)
                        .setContent(content)
                        .setFrom(ChannelLoginContext.getInstance().getLoginId(ch).toString())
                        .build();

                NetMsg netMsg = new NetMsg()
                        .setCmdId(CommandId.SRV_PUSH)
                        .body(messagePush.toByteArray());

                for (Channel ch : topicQueue) {
                    if (this.ch != ch) {
                        ch.writeAndFlush(netMsg);
                    }
                }
            });
        }
    }
}
